package mq

import (
	"context"

	"github.com/lab-online/pkg/logger"
	amqp "github.com/rabbitmq/amqp091-go"
)

type Delivery = amqp.Delivery

type ConsumerConfig struct {
	Ctx       context.Context
	Queue     string
	Consumer  string
	AutoAck   bool
	Exclusive bool
	NoLocal   bool
	NoWait    bool
	Args      amqp.Table
}

// RegisterConsumer 注册消费者, 创建新 channel (需自行控制 channel 关闭或传ctx)并尝试 declare queue, 错误转为已定义错误
func (mq *MessageQueue) RegisterConsumer(
	queueConfig QueueDeclareConfig, consumeConfig ConsumerConfig, consumer func(*amqp.Delivery),
) (*amqp.Channel, error) {
	if channel, _, err := mq.DeclareQueue(queueConfig); err != nil {
		return nil, err
	} else {
		return channel, mq.RegisterDeclaredQueueConsumer(channel, consumeConfig, consumer)
	}
}

// RegisterDeclaredQueueConsumer 注册已声明的队列消费者, 错误转为已定义错误
// 若传入的 ctx 为 nil, 则使用 channel.Consume, 否则使用 channel.ConsumeWithContext
func (mq *MessageQueue) RegisterDeclaredQueueConsumer(
	channel *amqp.Channel, config ConsumerConfig, consumer func(*amqp.Delivery),
) error {
	var msgs <-chan amqp.Delivery
	var err error

	if config.Ctx == nil {
		msgs, err = channel.Consume(
			config.Queue, config.Consumer, config.AutoAck, config.Exclusive,
			config.NoLocal, config.NoWait, config.Args,
		)
	} else {
		msgs, err = channel.ConsumeWithContext(
			config.Ctx, config.Queue, config.Consumer, config.AutoAck,
			config.Exclusive, config.NoLocal, config.NoWait, config.Args,
		)
	}

	if err != nil {
		return ErrConsume
	}
	go UseConsumer(config.Queue, msgs, consumer)
	logger.Info("consumer registered:", config.Queue)
	return nil
}

// UseConsumer 从 msgs 中取出消息并消费, 直到 msgs 关闭, 应在 goroutine 中使用
func UseConsumer(queue string, msgs <-chan amqp.Delivery, consumer func(*Delivery)) {
	for d, ok := <-msgs; ok; d, ok = <-msgs {
		consumer(&d)
	}
	logger.Info("consumer closed:", queue)
}
